package cn.colins.raft.node;

import cn.colins.raft.RaftNode;
import cn.colins.raft.config.ServerConfig;
import cn.colins.raft.enity.*;
import cn.colins.raft.enums.NodeStatusEnums;
import cn.colins.raft.exception.RaftRunException;
import cn.colins.raft.rpc.domin.RpcRemoteMsg;
import cn.colins.raft.rpc.session.RpcSession;
import cn.colins.raft.rpc.session.RpcSessionFactory;
import cn.colins.raft.state.StateMachines;
import cn.colins.raft.task.ElectionTask;
import cn.colins.raft.task.HeartBeatTask;
import cn.colins.raft.utils.ThreadPoolUtils;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson.JSONObject;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @Author czl
 * @Description 节点服务
 **/
public class RaftNodeService {

    private static final Logger log = LoggerFactory.getLogger(RaftNodeService.class);

    // 心跳间隔时间
    private final static long INTERVAL_TIME = 1500L;

    private static Map<NodeStatusEnums, RaftNode> raftNodeMap = new ConcurrentHashMap<>(8);

    static {
        raftNodeMap.put(NodeStatusEnums.LEADER, new LeaderRaftNode());
        raftNodeMap.put(NodeStatusEnums.CANDIDATE, new CandidateRaftNode());
        raftNodeMap.put(NodeStatusEnums.FOLLOW, new FollowRaftNode());
    }

    /**
     * 节点信息初始化
     */
    public static void raftNodeInit(ServerConfig self, List<ServerConfig> clusterConfig) {
        RaftNodeInfo.getInstance().setSelf(self);
        RaftNodeInfo.getInstance().setClusterConfig(clusterConfig);
        RaftNodeInfo.getInstance().setCurrentNodeStatus(NodeStatusEnums.FOLLOW);
        createElectionTask();
    }

    /**
     * 发送心跳
     */
    public  static void sendHeartBeat() {
        RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());
        boolean result = false;
        try {
            result = raftNode.callHeartBeatRequest(RaftNodeInfo.getInstance().getClusterConfig());
        } catch (ExecutionException e) {
            log.debug(" {}: 完了，作为leader发送心跳失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);
        } catch (InterruptedException e) {
            log.debug(" {}: 完了，作为leader发送心跳失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);
        }
        if (!result) {
            // 代表心跳失败了，状态已经变更了
            // 需要停止心跳，开启心跳检测
            heartBeatTestDestroy();
            createElectionTask();
        }
    }

    /**
     * 心跳处理
     */
    public static void heartBeatHandler(HeartBeatRequest request, Channel channel) {
        ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {
            RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());
            channel.writeAndFlush(new RpcRemoteMsg<HeartBeatResult>(raftNode.heartBeatHandler(request)));
            // 收到了心跳，所以就要停止当前心跳的检测，然后重新开启一个检测任务
            createElectionTask();
        });
    }

    // 命令合规性校验 目前就get set 随便校验一下
    public static boolean commandCheck(String command) {
        if (command == null || !"SET_GET".contains(command.split(" ")[0]) || command.split(" ").length < 2) {
            return false;
        }
        return true;
    }

    /**
     * 客户端的请求，以KV为例 就是set命令 , 这里请求返回就简陋一点
     */
    public static void clientRequestHandler(ClientRequest request, Channel channel) {
        ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {
            ClientResponse clientResponse = ClientResponse.builder().build();
            clientResponse.setRequestId(request.getRequestId());
            if (!commandCheck(request.getCommand())) {
                clientResponse.setCode(401);
                clientResponse.setMsg("命令格式不正确");
                channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(clientResponse));
                return;
            }
            // 只有set命令才需要发送日志，get命令直接取数据就行了
            String[] command = request.getCommand().split(" ");
            if (command[0].equals("SET")) {
                RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());
                try {
                    channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(raftNode.clientRequestHandler(request, RaftNodeInfo.getInstance().getClusterConfig())));
                    return;
                } catch (ExecutionException e) {
                    log.debug(" {}: 日志提交失败了:{}", request.getCommand(), e.getMessage(), e);
                    clientResponse.setCode(500);
                    clientResponse.setMsg(e.getMessage());
                    channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(clientResponse));
                } catch (InterruptedException e) {
                    log.debug(" {}: 日志提交失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);
                    clientResponse.setCode(500);
                    clientResponse.setMsg(e.getMessage());
                    channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(clientResponse));
                }
            } else {
                // get命令直接取值
                clientResponse.setCode(200);
                clientResponse.setData(RaftNodeInfo.getInstance().getLogManage().getDataByKey(command[1]));
                channel.writeAndFlush(new RpcRemoteMsg<ClientResponse>(clientResponse));
            }
        });
    }


    /**
     * Log预提交请求
     */
    public static void logPreCommitHandler(AppendEntriesPreCommit request, Channel channel) {
        ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {
            RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());
            channel.writeAndFlush(new RpcRemoteMsg<AppendEntriesPreCommitResult>(raftNode.logPreCommitHandler(request)));
        });
    }

    /**
     * Log提交请求
     */
    public static void logCommitHandler(AppendEntriesCommit request) {
        ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {
            RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());
            raftNode.logCommitHandler(request);
            // 收到了日志，所以就要停止当前心跳的检测，然后重新开启一个检测任务
            createElectionTask();
        });
    }

    /**
     * 发起投票
     */
    public  static void sendCallVote() {
        RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());
        boolean result = false;
        try {
            result = raftNode.callVoteRequest(RaftNodeInfo.getInstance().getClusterConfig());
        } catch (ExecutionException e) {
            StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);
            log.debug(" {}: 完了，作为candidate发起投票失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);
        } catch (InterruptedException e) {
            StateMachines.becomeFollow(RaftNodeInfo.getInstance().getCurrentTerm(), null, null);
            log.debug(" {}: 完了，作为candidate发起投票失败了:{}", RaftNodeInfo.getInstance().getSelf().toString(), e.getMessage(), e);
        }
        if (!result) {
            // 代表发起投票失败了，状态已经变更了
            // 需要重新开启一个检测任务
            createElectionTask();
            return;
        }
        // 投票成功了 需要开启心跳任务
        createHearBeatTask();
    }

    /**
     * 发起投票请求处理
     */
    public  static void callVoteHandler(RequestVoteRPC requestVoteRPC, Channel channel) {
        ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {
            RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());
            channel.writeAndFlush(new RpcRemoteMsg<RequestVoteResult>(raftNode.voteRequestHandler(requestVoteRPC)));
        });
    }

    /**
     * 发起Log拉取请求
     */
    public static LogIndexPullResult sendLogPullRequest(List<Long> pullLogIndex) {
        if (CollectionUtil.isNotEmpty(pullLogIndex)) {
            RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());
            return raftNode.sendLogPullRequest(pullLogIndex);
        }
        return null;
    }

    /**
     * 发起Log拉取请求处理
     */
    public static void logPullRequestHandler(LogIndexPull logIndexPull, Channel channel) {
        ThreadPoolUtils.nettyServerAsyncPool.execute(() -> {
            RaftNode raftNode = raftNodeMap.get(RaftNodeInfo.getInstance().getCurrentNodeStatus());
            log.debug("{}:拉取日志：{}",channel.remoteAddress(),JSONObject.toJSON(logIndexPull));
            channel.writeAndFlush(new RpcRemoteMsg<LogIndexPullResult>(raftNode.logPullRequestHandler(logIndexPull)));
        });
    }


    /**
     * 销毁并创建心跳检测任务
     */
    public static void createElectionTask() {
        long randomTime = getRandomTime();
        final long intervalTime = INTERVAL_TIME + randomTime;
        // 先销毁之前的
        electionTaskDestroy();
        //开启新的
        ScheduledFuture<?> schedule = ThreadPoolUtils.hearBeatAsyncPool.schedule(new ElectionTask(intervalTime), intervalTime, TimeUnit.MILLISECONDS);
        RaftNodeInfo.getInstance().setElectionTask(schedule);
    }

    /**
     * 销毁并创建心跳任务
     */
    public static void createHearBeatTask() {
        // 先销毁之前的
        heartBeatTestDestroy();
        //开启新的
        ScheduledFuture<?> schedule = ThreadPoolUtils.hearBeatAsyncPool.scheduleAtFixedRate(new HeartBeatTask(), 0L, INTERVAL_TIME, TimeUnit.MILLISECONDS);
        RaftNodeInfo.getInstance().setElectionTask(schedule);
    }


    public static long getRandomTime() {
        // 要比心跳慢一点
        return RandomUtil.randomLong(250L, 1000L);
    }

    /**
     * 销毁心跳检测任务
     */
    public static void electionTaskDestroy() {
        if (null != RaftNodeInfo.getInstance().getElectionTask()) {
            RaftNodeInfo.getInstance().getElectionTask().cancel(true);
            RaftNodeInfo.getInstance().setElectionTask(null);
        }
    }

    /**
     * 销毁心跳任务
     */
    public static void heartBeatTestDestroy() {
        if (null != RaftNodeInfo.getInstance().getHeartBeatTask()) {
            RaftNodeInfo.getInstance().getHeartBeatTask().cancel(true);
            RaftNodeInfo.getInstance().setHeartBeatTask(null);
        }
    }

    public static void main(String[] args) {
        for(int i=0;i<20;i++){
            System.out.println(RandomUtil.randomLong(250L, 750L));
        }
    }
}
